//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// nested_loop_join_executor.cpp
//
// Identification: src/execution/nested_loop_join_executor.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

#include "execution/executors/nested_loop_join_executor.h"
#include "binder/table_ref/bound_join_ref.h"
#include "common/exception.h"
#include "type/value_factory.h"

namespace bustub {

NestedLoopJoinExecutor::NestedLoopJoinExecutor(ExecutorContext *exec_ctx, const NestedLoopJoinPlanNode *plan,
                                               std::unique_ptr<AbstractExecutor> &&left_executor,
                                               std::unique_ptr<AbstractExecutor> &&right_executor)
    : AbstractExecutor(exec_ctx),
      plan_(plan),
      left_executor_(std::move(left_executor)),
      right_executor_(std::move(right_executor)) {
  if (!(plan->GetJoinType() == JoinType::LEFT || plan->GetJoinType() == JoinType::INNER)) {
    // Note for 2023 Spring: You ONLY need to implement left join and inner join.
    throw bustub::NotImplementedException(fmt::format("join type {} not supported", plan->GetJoinType()));
  }
}

void NestedLoopJoinExecutor::Init() {
  // std::cout << plan_->ToString(true) << std::endl;
  left_executor_->Init();
  right_executor_->Init();
  RID lrid;
  left_executor_->Next(&left_tuple_, &lrid);
  left_no_match_ = true;
}

auto NestedLoopJoinExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  auto join_type = plan_->GetJoinType();
  if (left_tuple_.GetLength() == 0) {
    return false;
  }

  if (join_type == JoinType::INNER) {
    return JoinInner(tuple, rid);
  } else if (join_type == JoinType::LEFT) {
    return JoinLeft(tuple, rid);
  }
  return false;
}

auto NestedLoopJoinExecutor::JoinInner(Tuple *tuple, RID *rid) -> bool {
  const Schema &lschema = left_executor_->GetOutputSchema();
  const Schema &rschema = right_executor_->GetOutputSchema();
  Tuple right_tuple;
  RID lrid, rrid;
  Value joinable;

  do {
    bool r_has = right_executor_->Next(&right_tuple, &rrid);
    if (not r_has) {
      if (left_executor_->Next(&left_tuple_, &lrid)) {
        right_executor_->Init();
        right_executor_->Next(&right_tuple, &rrid);
      } else {
        return false;
      }
    }
    joinable = plan_->Predicate()->EvaluateJoin(&left_tuple_, lschema, &right_tuple, rschema);
  } while (not(joinable.IsNull() || joinable.GetAs<bool>()));

  GenerateGoodTuple(tuple, right_tuple);

  return true;
}

auto NestedLoopJoinExecutor::JoinLeft(Tuple *tuple, RID *rid) -> bool {
  const Schema &lschema = left_executor_->GetOutputSchema();
  const Schema &rschema = right_executor_->GetOutputSchema();
  Value joinable;
  Tuple right_tuple;
  RID lrid, rrid;

  do {
    bool r_has = right_executor_->Next(&right_tuple, &rrid);
    if (not r_has) {
      if (left_no_match_) {
        left_no_match_ = false;
        GenerateNullTuple(tuple);
        return true;
      }
      if (left_executor_->Next(&left_tuple_, &lrid)) {
        right_executor_->Init();
        if (not right_executor_->Next(&right_tuple, &rrid)) {
          GenerateNullTuple(tuple);
          left_no_match_ = false;
          return true;
        }
        left_no_match_ = true;
      } else {
        left_no_match_ = false;
        return false;
      }
    }
    joinable = plan_->Predicate()->EvaluateJoin(&left_tuple_, lschema, &right_tuple, rschema);
  } while (not(joinable.IsNull() || joinable.GetAs<bool>()));

  left_no_match_ = false;
  GenerateGoodTuple(tuple, right_tuple);

  return true;
}

void NestedLoopJoinExecutor::GenerateNullTuple(Tuple *tuple) {
  const Schema &lschema = left_executor_->GetOutputSchema();
  const Schema &rschema = right_executor_->GetOutputSchema();
  std::vector<Value> vals;
  vals.reserve(lschema.GetColumnCount() + rschema.GetColumnCount());
  for (uint32_t i = 0; i < lschema.GetColumnCount(); ++i) {
    vals.emplace_back(left_tuple_.GetValue(&lschema, i));
  }
  for (uint32_t i = 0; i < rschema.GetColumnCount(); ++i) {
    vals.emplace_back(ValueFactory::GetNullValueByType(rschema.GetColumn(i).GetType()));
  }
  Tuple ret(vals, &GetOutputSchema());
  *tuple = std::move(ret);
}

void NestedLoopJoinExecutor::GenerateGoodTuple(Tuple *tuple, const Tuple &right_tuple) {
  const Schema &lschema = left_executor_->GetOutputSchema();
  const Schema &rschema = right_executor_->GetOutputSchema();
  std::vector<Value> vals;
  vals.reserve(lschema.GetColumnCount() + rschema.GetColumnCount());
  for (uint32_t i = 0; i < lschema.GetColumnCount(); ++i) {
    vals.emplace_back(left_tuple_.GetValue(&lschema, i));
  }
  for (uint32_t i = 0; i < rschema.GetColumnCount(); ++i) {
    vals.emplace_back(right_tuple.GetValue(&rschema, i));
  }
  Tuple ret(vals, &GetOutputSchema());
  *tuple = std::move(ret);
}
}  // namespace bustub